package com.alison.datastream;

import lombok.Data;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @Author alison
 * @Date 2024/4/3 13:35
 * @Version 1.0
 * @Description
 */

public class FilterStream {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<Person> flintstones  = env.fromElements(new Person("jack", 35),
                new Person("tom", 33),
                new Person("fred", 31));

        //从socket中获取数据流数据
//        DataStreamSource<String> env1 = env.socketTextStream("localhost", 9999);
//        DataStreamSource<String> csvEnv = env.readTextFile("file://E:/data/(sample)user_result.csv");


        DataStream<Person> adults = flintstones.filter(new FilterFunction<Person>() {
            @Override
            public boolean filter(Person person) throws Exception {
                return person.age >= 18;
            }
        });
        adults.print();

        env.execute();
    }





    @Data
    public static class Person {
        String name;
        Integer age;

        public Person() {
        }

        public Person(String name, Integer age) {
            this.name = name;
            this.age = age;
        }

        @Override
        public String toString() {
            return "Person{" +
                    "name='" + name + '\'' +
                    ", age=" + age +
                    '}';
        }
    }
}
